package com.atguigu.gmall.realtime.utils

import java.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.Partition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis

import scala.collection.mutable

/**
 * 偏移量管理工具类
 */
object MyOffsetUtils {

  /**
   * 将偏移量写入到redis中
   * 偏移量的结构为： 消费者group + topic + partition  =>offset
   *
   *redis中如何存储offset
   *      类型 : set zset string (hash) list
   *      key : topic-group
   *      value : partition-offset
   *      写入api :hset
   *      读取api :hgetall
   *      是否过期：no
   *
   */
  def saveOffset(topic : String , group : String ,offsetRanges: Array[OffsetRange] ) = {
    if(offsetRanges != null && offsetRanges.length > 0){
      val offsetValues = new java.util.HashMap[String,String]()

      //处理偏移量信息
      for (offsetRange <- offsetRanges) {
        val partition: Int = offsetRange.partition
        val endOffset: Long = offsetRange.untilOffset
        offsetValues.put(partition.toString,endOffset.toString)
      }
      println("写入偏移量 : " + offsetValues)
      //存储到redis中
      val jedis: Jedis = MyRedisUtils.getJedis()
      val offSetKey : String = s"offset:$topic:$group"
      jedis.hset(offSetKey,offsetValues)

      MyRedisUtils.close(jedis)
    }
  }

  //将偏移量读取出来
  def readOffset(topic : String , group : String): Map[TopicPartition,Long] = {
    val offsets: mutable.Map[TopicPartition, Long] = mutable.Map[TopicPartition,Long]()
    val jedis: Jedis = MyRedisUtils.getJedis()

    val offSetKey : String = s"offset:$topic:$group"
    val partitionOffset: util.Map[String, String] = jedis.hgetAll(offSetKey)
    println("读取到offset : " + partitionOffset)

    if(partitionOffset != null && partitionOffset.size() > 0){
      import scala.collection.JavaConverters._
      for ((partition,offset) <- partitionOffset.asScala) {
        val topicPartition = new TopicPartition(topic,partition.toInt)
        offsets.put(topicPartition,offset.toLong)
      }
    }
    
    MyRedisUtils.close(jedis)
    offsets.toMap
  }

}
